package cn.lecosa.util


import java.util
import com.typesafe.config.{Config, ConfigFactory}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.types.{DataType, StructField, StructType, _}
import scalikejdbc._
import scala.collection.JavaConversions._
import scala.collection.mutable.ListBuffer
import org.apache.spark.SparkConf

/**
  * Demo class
  *
  * @author administrator
  * @date 2019/1/2
  */
object BzkProperty extends Serializable {

  DBUtils.getDbConnection()
  private val config: Config = ConfigFactory.load()


  
//  下划线_代表的是某一类型的默认值。
//对于Int来说，它是0。
//对于Double来说，它是0.0
//对于引用类型，它是null。

  // 原始库名
  var yskm: String = _

  //原始表名称
  var ysbm: String = _

  // 需要查询的字段
  private var selectColumns: String = _

  // 数据项映射表
  private var sjxList: List[Sjxysb] = _

  //hdfs上的基础路径
  private val baseHafsPath: String = config.getString("spark.bzh.hdfs.basePath")

  def main(args: Array[String]): Unit = {
//        val querySql = getQuerySql()
       val sparkConfPrefix = this.getClass.getSimpleName.split("\\$")(0) + ".SparkConf"
    println(sparkConfPrefix)
    
        val sparkConf = new SparkConf();
       sparkConf.set("Hive2HiveSpark.SparkConf.spark.akka.frameSize","512")
       sparkConf.getAll.foreach(println)
       
  }
  /**
    * 获取组合主键
    */
  def getLasttime(): String = {
    checkNotNullAndInit
    //    val strings = sjxList.filter(x => {
    //      x.bzhsjlx != null && "2".equals(x.bzhsjxlx)
    //    }).map(_.bzhsjxmc)
    val strings = sjxList.map(_.bzhsjxmc)
    if (strings.nonEmpty) {
      return strings(0)
    } else {
      return "ml_qszxsj"
    }
  }

  /**
    * 获取组合主键
    */
  def getPrimaryKey() = {
    checkNotNullAndInit
    sjxList.map(_.bzhsjxmc)

  }

  /**
    * 获取查询语句
    */
  def getQuerySql() = {
    checkNotNullAndInit
    val selectColumns = getSelectColumns
    val bmysb = getBmys().get
    val where = getWhere()
    s"select ${selectColumns} from ${bmysb.yskm}.${ysbm}    ${where}"
  }

  /**
    * 增加分区sql
    */
  def addPartition() = {
    checkNotNullAndInit
    val bmysb = getBmys().get
    if ("1".equals(bmysb.sfzl)) {
      s"alter table ${bmysb.bzhkm}.${bmysb.bzhbm} add if not exists partition(${DateUtil.getdpDateStr()})"
    } else {
      ""
    }
  }

  /**
    * 填充原始数据项的 datatype信息
    *
    * @return
    */
  def fillSjxlist(schema: Seq[(String, DataType)]) = {
    checkNotNullAndInit
    sjxList.map(x => {
      schema.foreach(y => {
        if (x.yssjxmc.equalsIgnoreCase(y._1)) {
          x.schemaDatatype = y._2
        }
      })
    })
  }

  /**
    * 获取原始数据项，标准化数据项
    *
    * @return
    */
  def getSchema = {
    checkNotNullAndInit
    val sjxList_tmp = new util.ArrayList[Sjxysb]
    sjxList.foreach(r => {
      if (StringUtils.isNotEmpty(r.nrbzh) && r.nrbzh.contains("ysdmToBzhdm")) {
        sjxList_tmp.add(r)
        sjxList_tmp.add(Sjxysb("", r.yssjxmc, "string", "", r.bzhsjxmc + "_name",
          r.bzhzwms, "string", "", "", 15, 0, null))
      } else {
        sjxList_tmp.add(r)
      }

    })
    val structType = StructType(sjxList_tmp.map(x => {
      StructField(x.bzhsjxmc, x.bzhsjlx.toLowerCase match {
        case "string" => StringType
        case "long" => LongType
        case "bigint" => LongType
        case "double" => DoubleType
        case "int" => IntegerType
        case _ if (x.bzhsjlx.toLowerCase.contains("decimal")) => createDecaimalType(x)
        case _ => StringType
      })
    }))

    structType.add("mlsfcg", IntegerType)
      .add(StructField("mlsystemtime", LongType))
  }

  /**
    * 创建DecaimalType
    *
    * @param bzhsjlx
    * @return
    */
  def createDecaimalType(sjxys: Sjxysb) = {
    val precision = sjxys.colLength match {
      case _ if (sjxys.colLength > 0) => sjxys.colLength
      case _ => 15
    }
    val scale = sjxys.colScale match {
      case _ if (sjxys.colScale > 0) => sjxys.colScale
      case _ => 0
    }
    DecimalType(precision, scale)
  }

  def createDecaima(sjxys: Sjxysb) = {
    val precision = sjxys.colLength match {
      case _ if (sjxys.colLength > 0) => sjxys.colLength
      case _ => 15
    }
    val scale = sjxys.colScale match {
      case _ if (sjxys.colScale > 0) => sjxys.colScale
      case _ => 0
    }
    "decimal(" + precision + "," + scale + ")"
  }

  /**
    * 获取原始数据项，标准化数据项
    *
    * @return
    */
  def getsjxList = {
    checkNotNullAndInit
    sjxList.foreach(x => {
      if (x.schemaDatatype == null) {
        println("datatype 没有完全填充，程序退出")
        System.exit(0)
      }
    })
    sjxList
  }

  /**
    * 获取spark保存方式
    *
    * @return
    */
  def getSaveMode = {
    checkNotNullAndInit
    val maybeModel = getBmys()
    if (maybeModel.isEmpty) {
      println("表明映射表中无该表，程序退出")
      System.exit(1)
    }

    val bmysbeModel = maybeModel.get
    val sfzl = bmysbeModel.sfzl

    // 1 是增量更新 0 不更新的，2 是全量更新
    if ("1".equals(sfzl)) SaveMode.Append else SaveMode.Overwrite

  }

  /**
    * 获取spark保存的路径
    */
  def getSavePath = {
    checkNotNullAndInit
    val maybeModel = getBmys()
    if (maybeModel.isEmpty) {
      println("表明映射表中无该表，程序退出")
      System.exit(1)
    }

    val bmysbeModel = maybeModel.get
    val sfzl = bmysbeModel.sfzl

    // 1 是增量更新 0 不更新的，2 是全量更新
    if ("1".equals(sfzl)) {
      baseHafsPath + bmysbeModel.bzhbm.toLowerCase + "/" + DateUtil.getdpDateStr()
    } else {
      baseHafsPath + bmysbeModel.bzhbm.toLowerCase
    }

  }

  /**
    * 检查 原始库名、原始表名、sjxList 不为空，并且初始化
    */
  private def checkNotNullAndInit = {
    if (ysbm == null) {
      println("原始表名不能为空，程序退出")
      System.exit(0)
    }
    if (sjxList == null || sjxList.isEmpty) {
      init()
    }
    if (sjxList.isEmpty) {
      println("没有映射关系，程序退出")
      System.exit(0)
    }
  }

  /**
    * 获取查询条件
    *
    * @param sjxysb
    * @return
    */
  def getWhere() = {
    val bmysb = getBmys().get
    val lastRunstatus = DB.readOnly { implicit session =>
      sql"select zlzd, zlzdz from s_xsparkyxzt where ysbm = ? and bzhbm = ?"
        .bind(bmysb.ysbm.toUpperCase, bmysb.bzhbm.toUpperCase).map(rs => {
        (rs.string(1), rs.string(2))
      }).single().apply()
    }

    if (lastRunstatus.isEmpty) {
      "  where 1=1 "
    } else {
      val runstatus = lastRunstatus.get
      if ("wzl".equals(runstatus._2)) {
        " where 1=2"
      } else if ("全量更新".equals(runstatus._2)) {
        "  where 1=1 "
      } else {
        " where 1=1 and " + runstatus._1.toLowerCase + " > '" + runstatus._2 + "'"
      }
    }
  }

  /**
    * 获取创建hive表的建表语句
    *
    * @param sjxList
    * @param path
    * @return
    */
  def createHiveDDL(): String = {
    checkNotNullAndInit
    val bmysModel = getBmys()
    val bmysb = bmysModel.get
    val sb = new StringBuilder(s"CREATE EXTERNAL TABLE IF NOT EXISTS ${bmysb.bzhkm}.").append(bmysb.bzhbm).append("(").append("\n")

    sjxList.foreach(x => {
      if ("DECIMAL".equalsIgnoreCase(x.bzhsjlx)) {
        sb.append(x.bzhsjxmc).append(" ").append(createDecaima(x)).append(" COMMENT '").append(x.bzhzwms).append("',").append("\n")
      } else {
        sb.append(x.bzhsjxmc).append(" ").append(x.bzhsjlx).append(" COMMENT '").append(x.bzhzwms).append("',").append("\n")
      }

      //内容码值转换添加一个中文描述字段
      if (StringUtils.isNotEmpty(x.nrbzh) && x.nrbzh.contains("ysdmToBzhdm")) {
        sb.append(x.bzhsjxmc + "_mc").append(" ").append("string").append(" COMMENT '").append(x.bzhzwms).append("',").append("\n")
      }
    })

    sb.append("mlsfcg").append(" ").append(" int").append(" COMMENT '").append("是否处理成功").append("',").append("\n")
    sb.append("mlsystemtime").append(" ").append(" bigint").append(" COMMENT '").append("标准化处理时间").append("',").append("\n")


    sb.delete(sb.lastIndexOf(","), sb.length).append(") \n")
      .append(" COMMENT  '").append(bmysb.bzhzwms).append("'\n")
    if ("1".equalsIgnoreCase(bmysb.sfzl)) {

      var fqzd = bmysb.fqzd
      if ("".equals(StringUtils.trimToEmpty(fqzd))) {
        fqzd = "dp"
      }
      sb.append(" PARTITIONED BY (`").append(fqzd.toLowerCase).append("` string  COMMENT '分区字段' )").append("\n")
    }
    sb.append("ROW FORMAT DELIMITED NULL DEFINED AS ''").append("\n")
    //    val str = sb.append("STORED AS PARQUET LOCATION ").append("'").append(baseHafsPath + bmysb.bzhbm.toLowerCase).append("'").toString()
    //TEXT格式hive表
    val str = sb.append("STORED AS TEXTFILE LOCATION ").append("'").append(baseHafsPath + bmysb.bzhbm.toLowerCase).append("'").toString()
    //    val str = sb.toString()
    str
  }

  def alterTableNull(): String = {
    val bmysModel = getBmys().get
    val alterSQL = "alter table " + bmysModel.bzhkm + '.' + bmysModel.bzhbm + " set serdeproperties('serialization.null.format' = ' ')"
    alterSQL
  }

  /**
    * 获取需要查询的字段
    *
    * @return
    */
  private def getSelectColumns = {
    checkNotNullAndInit
    val bmysb = getBmys().get
    var str = sjxList.map(_.yssjxmc).mkString(",")
    //    if("1".equals(bmysb.sfzl)){
    //      str = str + s",${bmysb.zlzd}"
    //    }
    str

  }

  /**
    * 根据原始表名和标准化表明，获取表明映射表信息
    *
    * @param ysbm
    * @param bzhbm
    * @return
    */
  def getBmys() = {
    checkNotNullAndInit
    val sjxysb = sjxList(0)

    val bmysbModel = DB.readOnly { implicit session =>
      sql"select ysmmkj, ysbm,bzhmmkj, bzhbm, bzhzwms, sfzl, fqzd, zlzd from s_bmysb where ysbm = ? and bzhbm = ?"
        .bind(sjxysb.ysbm, sjxysb.bzhbm).map(rs => {
        BmysbModel(rs.string(1), rs.string(2), rs.string(3), rs.string(4),
          rs.string(5), rs.string(6), rs.string(7), rs.string(8))
      }).single().apply()
    }


    bmysbModel
  }

  /**
    * 初始化，生成数据项映射表关系
    */
  private def init(): Unit = {
    sjxList = DB.readOnly { implicit session => {
      sql"select t.ysbm,t.yssjxmc,t.yssjlx,t.bzhbm,t.bzhsjxmc,t.bzhzwms,t.bzhsjlx,t.sjybsf,t.nrbzh_el,t.std_col_length,t.std_col_scale from s_sjxmysb t where t.ysbm= ? order by t.std_col_position"
        .bind(ysbm).map(rs => {
        Sjxysb(StringUtils.trimToEmpty(rs.string(1)).toLowerCase,
          StringUtils.trimToEmpty(rs.string(2)).toLowerCase,
          rs.string(3),
          StringUtils.trimToEmpty(rs.string(4)).toLowerCase,
          StringUtils.trimToEmpty(rs.string(5)).toLowerCase,
          rs.string(6),
          rs.string(7),
          rs.string(8),
          rs.string(9),
          rs.int(10),
          rs.int(11),
          null
        )
      }).list().apply()
    }
    }
  }

}


/**
  * 数据项映射表及其在hive中的字段类型
  *
  * @param ysbm
  * @param yssjxmc
  * @param yssjlx
  * @param bzhbm
  * @param bzhsjxmc
  * @param bzhzwms
  * @param bzhsjlx
  * @param sjybsf
  * @param nrbzh
  * @param schemaDatatype
  */
case class Sjxysb(ysbm: String, yssjxmc: String, yssjlx: String, bzhbm: String, bzhsjxmc: String,
                  bzhzwms: String, bzhsjlx: String, sjybsf: String, nrbzh: String, colLength: Int, colScale: Int, var schemaDatatype: DataType)

/**
  * 表名映射表中的字段
  *
  * @param yskm
  * @param ysbm
  * @param bzhkm
  * @param bzhbm
  * @param bzhzwms
  * @param sfzl
  * @param fqzd
  * @param zlzd
  */
case class BmysbModel(yskm: String, ysbm: String, bzhkm: String, bzhbm: String, bzhzwms: String, sfzl: String, fqzd: String, zlzd: String)